package com.atguigu.gmall.realtime.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.util.DimUtilHbase;
import com.atguigu.gmall.realtime.util.HbaseUtil;
import com.atguigu.gmall.realtime.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

import java.io.IOException;

/**
 * @Author lzc
 * @Date 2023/6/1 18:02
 */
public class DimSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private Connection hbaseConn;
    private Jedis redisClient;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        hbaseConn = HbaseUtil.getHbaseConnection();
        
        redisClient = RedisUtil.getRedisClient();
    }
    
    @Override
    public void close() throws Exception {
        HbaseUtil.closeHbaseConnection(hbaseConn);
        
        redisClient.close();
    }
    
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value, Context context) throws Exception {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        
        String opType = data.getString("op_type");
        
        if ("delete".equals(opType)) { // 如果mysql 中delete了维度,则hbase 中的维度也应该删除
            del(data, tp);
        } else { // 如果 mysql 中 insert update 维度,则 hbase 中应该 put数据
            put(data, tp);  //
        }
        
        
        if ("delete".equals(opType) || "update".equals(opType)) { // 如果mysql 中delete了维度,则hbase 中的维度也应该删除
            redisClient.del(DimUtilHbase.getKey(tp.getSinkTable(), data.getString("id")));
        }
    }
    
    private void put(JSONObject data, TableProcess tp) throws IOException {
        String[] columns = tp.getSinkColumns().split(",");
        String[] values = new String[columns.length];
        for (int i = 0; i < columns.length; i++) {
            values[i] = data.getString(columns[i]); // 获取列对应的值
        }
        
        String rowKey = data.getString(tp.getSinkRowKey());
        HbaseUtil.putRow(hbaseConn,
                         "gmall",
                         tp.getSinkTable(),
                         rowKey,
                         tp.getSinkFamily(),
                         columns,
                         values
        );
        
    }
    
    private void del(JSONObject data, TableProcess tp) throws IOException {
        String rowKey = data.getString(tp.getSinkRowKey());
        HbaseUtil.delRow(hbaseConn, "gmall", tp.getSinkTable(), rowKey);
    }
}
